package com.atguigu.gmall.realtime.common.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.bean.TradeSkuOrderBean;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.HBaseUtil;
import com.atguigu.gmall.realtime.common.util.RedisUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

public abstract class DimRichMapFunction<T> extends RichMapFunction<T,T> implements DimFunction<T>{

    private Connection connection;
    private Jedis jedis;

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = HBaseUtil.getConnection();
        jedis = RedisUtil.getJedis();
    }

    @Override
    public void close() throws Exception {
        HBaseUtil.closeConnection(connection);
        RedisUtil.closeJedis(jedis);
    }

    @Override
    public  T map(T bean) throws Exception {
        String rowKey = getRowKey(bean);
        String tableName = getTableName();
        // 优先从Redis中读取维度数据
        JSONObject dimJsonObj = RedisUtil.readDim(jedis, tableName, rowKey);
        if (dimJsonObj == null){
            // 从HBase中读取数据
            dimJsonObj =
                    HBaseUtil.getRow(connection, Constant.HBASE_NAMESPACE, tableName, rowKey, JSONObject.class, false);

            // 读取到写入到Redis
            if (dimJsonObj.size() > 0){
                RedisUtil.writeDim(jedis, tableName, rowKey, dimJsonObj);
            }
        }
        if (dimJsonObj.size() > 0) {
            //关联维度
            addDims(bean, dimJsonObj);
        }
        return bean;
    }

}
